﻿namespace DataIntegration.Messaging.Consumer.ActiveMQ
{
    using System;
    using System.Reflection;
    using Apache.NMS;
    using Apache.NMS.Util;
    using log4net;
    using Utils.Lifecycle.Base;

    public class ActiveMQConsumer : BaseLifecycle, IConsumer
    {
        public event Action<string, string> MessageReceived;

        private static readonly ILog Logger = LogManager.GetLogger(MethodBase.GetCurrentMethod().ReflectedType);

        private IConnection connection;

        private ISession session;

        private IMessageConsumer consumer;

        private readonly IConnectionFactory connectionFactory;

        public ActiveMQConsumer(IConnectionFactory connectionFactory)
        {
            Logger.DebugFormat("Creating ActiveMQConsumer");
            this.connectionFactory = connectionFactory;
        }

        protected override void Setup()
        {
            Logger.DebugFormat("Starting ActiveMQConsumer");
            this.connection = this.connectionFactory.CreateConnection();
            this.session = this.connection.CreateSession();
            this.connection.Start();
        }

        public override void Dispose()
        {
            this.Dispose(this.consumer);
            this.Dispose(this.session);
            this.Dispose(this.connection);
        }

        public void Consume(string topic)
        {
            if (this.MessageReceived == null)
            {
                throw new ArgumentException("No handler is bound to MessageReceived");
            }

            this.consumer = this.session.CreateConsumer(SessionUtil.GetTopic(this.session, topic));
            this.consumer.Listener += this.InvokeMessageReceived;
        }

        public void ConsumeDurable(string topic)
        {
            if (this.MessageReceived == null)
            {
                throw new ArgumentException("No handler is bound to MessageReceived");
            }

            ITopic destination = SessionUtil.GetTopic(this.session, topic);
            this.consumer = this.session.CreateDurableConsumer(
                destination,
                destination.TopicName,
                null,
                false);
            this.consumer.Listener += this.InvokeMessageReceived;
        }

        private void InvokeMessageReceived(IMessage message)
        {
            Logger.InfoFormat("Received message with id=[{0}] on destination=[{1}]", message.NMSCorrelationID, message.NMSDestination);

            ITextMessage textMessage = message as ITextMessage;

            if (textMessage != null)
            {
                if (this.MessageReceived != null)
                {
                    this.MessageReceived(message.NMSCorrelationID, textMessage.Text);
                }
            }
        }
    }
}